package com.paic.icore.test.demo.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * describe: 多生产者，单消费者
 *
 * @author huxingjie
 * @date 2019/03/22
 * @since 3.4
 */
public class MultiProductorLongEventMain {
    private static final Translator TRANSLATOR = new Translator();

    public static void main(String[] args) throws Exception {
        // Executor that will be used to construct new threads for consumers
        Executor executor = Executors.newCachedThreadPool();
        ThreadFactory threadFactory = new ThreadFactory() {
            private final AtomicInteger atomicInteger = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(null, r, "disruptor-thread " + atomicInteger.getAndIncrement());
            }
        };
        // Specify the size of the ring buffer, must be power of 2.
        int bufferSize = 1024;
        // Construct the Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(LongEvent::new, bufferSize, threadFactory, ProducerType.MULTI, new BlockingWaitStrategy());
        // Connect the handler
        disruptor.handleEventsWith(new LongEventHandler());
        // Start the Disruptor, starts all threads running
        disruptor.start();
        // Get the ring buffer from the Disruptor to be used for publishing.
        new Thread() {
            @Override
            public void run() {
                RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

                ByteBuffer bb = ByteBuffer.allocate(8);
                for (long l = 0; l < 10; l++) {
                    onData(ringBuffer, bb, l);
                }
            }
        }.start();
        new Thread() {
            @Override
            public void run() {
                RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

                ByteBuffer bb = ByteBuffer.allocate(8);
                for (long l = 10; l < 20; l++) {
                    onData(ringBuffer, bb, l);
                }
            }
        }.start();
        new Thread() {
            @Override
            public void run() {
                RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

                ByteBuffer bb = ByteBuffer.allocate(8);
                for (long l = 20; l < 30; l++) {
                    onData(ringBuffer, bb, l);
                }
            }
        }.start();

    }

    private static void onData(RingBuffer<LongEvent> ringBuffer, ByteBuffer bb, long l) {
        bb.putLong(0, l);
        ringBuffer.publishEvent(TRANSLATOR, bb);

        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

    